package com.matrix.im.netty.handler;

import com.matrix.api.user.UserApi;
import com.matrix.api.user.vo.UserInfoVO;
import com.matrix.common.enums.UserTypeEnum;
import com.matrix.im.application.IMAppService;
import com.matrix.im.config.ImThreadPoolConfig;
import com.matrix.im.interfaces.dto.SendMessageDTO;
import com.matrix.im.netty.session.SessionManager;
import com.matrix.im.protocol.MessageProto;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.Executor;

/**
 * @author 有点甜
 * @since 2025/8/6
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class MessageHandler extends SimpleChannelInboundHandler<MessageProto.MessagePackage> {

    private final IMAppService imAppService;
    private final Executor businessExecutor;

    @DubboReference(check = false)
    private UserApi userApi;

    public MessageHandler(IMAppService imAppService,
                          @Qualifier(ImThreadPoolConfig.IM_BUSINESS_EXECUTOR) Executor businessExecutor) {
        this.imAppService = imAppService;
        this.businessExecutor = businessExecutor;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProto.MessagePackage msg) {
        Long fromUserId = ctx.channel().attr(SessionManager.USER_ID_KEY).get();
        if (fromUserId == null) {
            log.warn("Message received from unauthenticated channel, dropping message.");
            return;
        }
        businessExecutor.execute(() -> {
            try {
                switch (msg.getPackageType()) {
                    case PRIVATE_MESSAGE_REQUEST:
                        handlePrivateMessage(ctx, msg.getPrivateMessageRequest(), fromUserId);
                        break;
                    case HEARTBEAT_REQUEST:
                        handleHeartbeat(ctx, fromUserId);
                        break;
                    default:
                        log.warn("User {} sent unhandled package type: {}", fromUserId, msg.getPackageType());
                        break;
                }
            } catch (Exception e) {
                log.error("Error in business executor for user {}: {}", fromUserId, e.getMessage(), e);
                String clientMessageId = getClientMessageId(msg);
                if (clientMessageId != null) {
                    sendErrorAckInEventLoop(ctx, clientMessageId, "服务器内部错误");
                }
            }
        });
    }

    private void handlePrivateMessage(ChannelHandlerContext ctx, MessageProto.PrivateMessageRequest request, Long fromUserId) {
        UserTypeEnum fromUserType = ctx.channel().attr(SessionManager.USER_TYPE_KEY).get();
        Long toUserId = request.getToUserId();

        if (fromUserType == UserTypeEnum.MERCHANT) {
            try {
                List<UserInfoVO> users = userApi.getUsersByIds(List.of(toUserId));
                if (users.isEmpty() || users.get(0).getUserType() != UserTypeEnum.USER) {
                    log.warn("Permission denied: Merchant {} attempted to chat with non-user {}", fromUserId, toUserId);
                    sendErrorAckInEventLoop(ctx, request.getClientMessageId(), "商家只能与普通用户进行沟通");
                    return;
                }
            } catch (Exception e) {
                log.error("Failed to verify user type for chat. from: {}, to: {}", fromUserId, toUserId, e);
                sendErrorAckInEventLoop(ctx, request.getClientMessageId(), "内部服务错误,消息发送失败");
                return;
            }
        }

        MessageProto.MessageBody body = request.getBody();

        SendMessageDTO dto = new SendMessageDTO();
        dto.setToUserId(request.getToUserId());
        dto.setClientMessageId(request.getClientMessageId());

        SendMessageDTO.MessageBodyDTO bodyDto = new SendMessageDTO.MessageBodyDTO();
        bodyDto.setType(body.getType().name());
        bodyDto.setContent(body.getContent());
        dto.setBody(bodyDto);

        try {
            imAppService.sendMessage(fromUserId, dto);
            sendSuccessAckInEventLoop(ctx, request.getClientMessageId());
        } catch (Exception e) {
            log.error("Error processing private message from user {}", fromUserId, e);
            sendErrorAckInEventLoop(ctx, request.getClientMessageId(), e.getMessage());
        }
    }

    private void handleHeartbeat(ChannelHandlerContext ctx, Long fromUserId) {
        log.debug("Received heartbeat from user {}", fromUserId);
    }


    private void sendSuccessAckInEventLoop(ChannelHandlerContext ctx, String clientMessageId) {
        MessageProto.MessageAck ack = MessageProto.MessageAck.newBuilder()
                .setClientMessageId(clientMessageId)
                .setSuccess(true)
                .setMessage("Message received by server.")
                .build();
        MessageProto.MessagePackage pkg = buildAckPackage(ack);

        ctx.channel().eventLoop().execute(() -> ctx.channel().writeAndFlush(pkg));
    }

    private void sendErrorAckInEventLoop(ChannelHandlerContext ctx, String clientMessageId, String errorMessage) {
        MessageProto.MessageAck ack = MessageProto.MessageAck.newBuilder()
                .setClientMessageId(clientMessageId)
                .setSuccess(false)
                .setMessage(errorMessage)
                .build();
        MessageProto.MessagePackage pkg = buildAckPackage(ack);

        ctx.channel().eventLoop().execute(() -> ctx.channel().writeAndFlush(pkg));
    }

    private String getClientMessageId(MessageProto.MessagePackage msg) {
        if (msg.hasPrivateMessageRequest()) {
            return msg.getPrivateMessageRequest().getClientMessageId();
        }
        return null;
    }

    private MessageProto.MessagePackage buildAckPackage(MessageProto.MessageAck ack) {
        return MessageProto.MessagePackage.newBuilder()
                .setPackageType(MessageProto.PackageType.MESSAGE_ACK)
                .setMessageAck(ack)
                .build();
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("An unhandled error occurred in MessageHandler pipeline for channel {}: {}", ctx.channel().remoteAddress(), cause.getMessage(), cause);
        ctx.close();
    }
}
